package cn.admin.configcenter.registry.service.push;

import cn.admin.config.common.Revision;
import cn.admin.config.common.protocol.Swizzle;
import cn.admin.configcenter.registry.IOCFactory;
import cn.admin.configcenter.registry.common.info.ClientConnection;
import cn.admin.configcenter.registry.common.info.GroupId;
import cn.admin.configcenter.registry.common.task.Task;
import cn.admin.configcenter.registry.common.task.TaskProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class PushDelayTask extends Task {

    private static final Logger log = LoggerFactory.getLogger(PushDelayTask.class);
    static final String TASK_TYPE = PushDelayTask.class.getName();
    static final long PUSH_DELAY = 1000L;
    final GroupId group;
    final Set<ClientConnection> dest;
    volatile boolean isPushToAll;
    PushProfileTime pushProfileTime = new PushProfileTime();

    public PushDelayTask(GroupId group) {
        super(taskId(group),TASK_TYPE, PUSH_DELAY);
        this.pushProfileTime.createTime = System.currentTimeMillis();
        this.group = group;
        this.dest = null;
        this.isPushToAll = true;
    }

    public PushDelayTask(GroupId group,ClientConnection dest) {
        super(taskId(group),TASK_TYPE,PUSH_DELAY);
        this.pushProfileTime.createTime = System.currentTimeMillis();
        this.group = group;
        this.isPushToAll = false;
        this.dest = new HashSet<>();
        if (dest == null) {
            throw new IllegalArgumentException("subscribers is null");
        }
    }

    static String taskId(GroupId group) {
        return "PushTask[" + group.dataId + ", " + group.groupId + "]";
    }

    @Override
    protected boolean shouldProcess() {
        return super.shouldProcess();
    }

    @Override
    protected void merge(Task existTask) {
        PushDelayTask task = (PushDelayTask) existTask;
        if (!this.isPushToAll && !task.isPushToAll) {
            this.dest.addAll(task.dest);
        } else {
            this.isPushToAll = true;
        }


    }

    static class PushDataTaskProcessor implements TaskProcessor {

        private static final int PUSH_THREAD = 2;
        private static final ExecutorService executePushTaskPool;
        private static final Semaphore semaphore;

        @Override
        public boolean process(Task task) throws Exception {

            PushDelayTask t = (PushDelayTask) task;
            t.pushProfileTime.createTime = System.currentTimeMillis();
            GroupId group = t.group;
            Revision revision;
            List<Swizzle> dataList;
            try {
                revision = group.revision();
                dataList = IOCFactory.getDataService().getPushData(group);
            } catch (Exception e) {
                PushDelayTask.log.error("########## get push data failed, "
                        + group + ", " + e.toString(), e);
                return false;
            }

            t.pushProfileTime.finishCountPushDataTime = System.currentTimeMillis();

            if (t.isPushToAll) {
                semaphore.acquire();
                executePushTaskPool.execute(new PushExecuteTask(group, dataList, revision,
                        t.pushProfileTime, semaphore));
            } else {
                if (t.dest != null) {
                    for (ClientConnection client : t.dest) {
                        semaphore.acquire();
                        executePushTaskPool.execute(new PushExecuteTask(group, dataList,
                                revision, client,
                                t.pushProfileTime, semaphore));
                    }
                }
            }

            return true;
        }

        static {
            executePushTaskPool = Executors.newFixedThreadPool(PUSH_THREAD,
                    r -> {
                        Thread t = new Thread(r);
                        t.setName("pushExecuteTask-Thread");
                        return t;
                    });
            semaphore = new Semaphore(2);
        }

    }
}
